package org.example.lookup_join;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkUserTableWithEventTimeDataGen {
    public static void main(String[] args) throws Exception {
        // 配置Flink Web UI端口
        Configuration config = new Configuration();
        config.setString("rest.bind-port", "8083"); // 设置Web UI端口为8083

        // 设置流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        env.setParallelism(1); // 设置并行度为1
        env.enableCheckpointing(5000); // 开启checkpoint，间隔5秒
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建Paimon目录
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println("执行SQL: " + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);

        // 在创建用户表之前添加datagen表创建语句
        String createDatagenTableSQL = "CREATE TABLE IF NOT EXISTS `datagen_t_user_table_with_eventtime` (\n" +
                "    user_id BIGINT COMMENT '用户ID',\n" +
                "    user_name STRING COMMENT '用户名',\n" +
                "    user_email STRING COMMENT '用户邮箱',\n" +
                "    user_address STRING COMMENT '用户地址',\n" +
                "    update_time TIMESTAMP(3) COMMENT '用户信息更新时间',\n" +
                "    WATERMARK FOR update_time AS update_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "    'connector' = 'datagen',\n" +
                "    'rows-per-second' = '200',\n" +
                "    'number-of-rows' = '100',\n" +
                "    'fields.user_id.min' = '1',\n" +
                "    'fields.user_id.max' = '100',\n" +
                "    'fields.user_name.length' = '5',\n" +
                "    'fields.user_email.length' = '10',\n" +
                "    'fields.user_address.length' = '15'\n" +
                ");";
        System.out.println("执行SQL: " + createDatagenTableSQL);
        tableEnv.executeSql(createDatagenTableSQL);
        
        // DROP掉之前创建的PAIMON表
        String dropUserTableSQL = "DROP TABLE IF EXISTS `paimon`.`default`.`t_user_table_with_eventtime`;";
        System.out.println("执行SQL: " + dropUserTableSQL);
        tableEnv.executeSql(dropUserTableSQL);  

        // 创建带事件时间的用户表（如果不存在）
        String createUserTableSQL = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`t_user_table_with_eventtime` (\n" +
                "    user_id BIGINT PRIMARY KEY NOT ENFORCED COMMENT '用户ID（主键）',\n" +
                "    user_name STRING COMMENT '用户名',\n" +
                "    user_email STRING COMMENT '用户邮箱',\n" +
                "    user_address STRING COMMENT '用户地址',\n" +
                "    update_time TIMESTAMP(3) COMMENT '用户信息更新时间',\n" +
                "    WATERMARK FOR update_time AS update_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "    'connector' = 'paimon',\n" +
                "    'bucket' = '-1'\n" +
                ");";
        System.out.println("执行SQL: " + createUserTableSQL);
        tableEnv.executeSql(createUserTableSQL);

        // 将datagen数据写入Paimon表
        String insertSQL = "INSERT INTO `paimon`.`default`.`t_user_table_with_eventtime`\n" +
                "SELECT * FROM `datagen_t_user_table_with_eventtime`;";
        System.out.println("执行SQL: " + insertSQL);
        tableEnv.executeSql(insertSQL);
    }
} 